package org.niit.stream

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
/*
  利用Spark Streaming 对 Kafka中的数据 进行消费

 */
object SparkStreaming_Kafka {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    ssc.sparkContext.setLogLevel("ERROR")

    //1.配置 连接Kafka的基本信息
    // kafka-console-consumer.sh --bootstrap-server node1:9092
    val kafkaParams = Map[String,Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "KP",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer].getName,
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer].getName //将主题中字节数组的数据解析为字符串
    )
    //2. 连接Kafka 获得Kafka中的数据
    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent, //官方推荐的本地策略
                                                    //主题
      ConsumerStrategies.Subscribe[String, String](Set("test"), kafkaParams)
    )

    //打印获取到信息
   val infoDs =  ds.map( line =>{
      val topic: String = line.topic() //通过数据获得当前的主题
      val partition: Int = line.partition()//通过数据获得当前的分区
      val offset: Long = line.offset()
      val value: String = line.value()

      val info = s"主题${topic},分区${partition},偏移量${offset},值${value}"
      info
    })

    infoDs.print()

    ssc.start()
    ssc.awaitTermination()
  }
}
